package cn.wangjie.spark.kafka.source

import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 使用Structured Streaming从Kafka实时读取数据，进行词频统计，将结果打印到控制台。
 */
object StructuredKafkaSource {

  def main(args: Array[String]): Unit = {
		
		// 创建SparkSession实例对象
		val spark: SparkSession = SparkSession
			.builder()
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			.master("local[2]")
			// 设置SparkSQL Shuffle时分区数目
			.config("spark.sql.shuffle.partitions", "2")
			.getOrCreate()
		// 导入隐式转换和函数库
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		
		// TODO: 1. 从Kafka消费数据
		val kafkaStreamDF: DataFrame = spark
			.readStream
			.format("kafka")
			.option("kafka.bootstrap.servers", "node1.itcast.cn:9092")
			.option("subscribe", "wordsTopic")
			.load()
		
		// TODO: 2. 获取每条消息Message
		val resultStreamDF: DataFrame = kafkaStreamDF
			// 将value转换为String字符串类型
			.selectExpr("CAST(value AS STRING)")
			// 将DataFrame转换为Dataset
			.as[String]
			// 过滤数据
			.filter(line => null != line && line.trim.length > 0)
			// 分割单词
			.flatMap(line => line.trim.split("\\s+"))
			// 按照单词分组，聚合
			.groupBy($"value").count()
		
		//  输出数据和启动启用
		val query: StreamingQuery = resultStreamDF
			.writeStream
			// 使用Update更新模式输出，当ResultTable中数据更新时，才将更新数据输出
			.outputMode(OutputMode.Update())
			.format("console")
			.option("numRows", "10")
			.option("truncate", "false")
			.start()
		query.awaitTermination() // 当流式应用运行以后，正常情况一直运行，除非人为或程序异常终止，等待结束
		query.stop()
		
	}
	
}
